package com.example.toy.core.commit;

import com.example.toy.core.utils.ToyPathDefine;
import com.example.toy.core.version.ToyVersion;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import lombok.SneakyThrows;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;

/**
 * DefaultToyCommitView
 *
 * @author shenbing
 * @date 2021/12/20
 */
public class FsOfflineRealtimeToyCommitView implements ToyCommitView {
  protected final Path basePath;
  protected final Path commitDirPath;

  @SneakyThrows
  public FsOfflineRealtimeToyCommitView(Path basePath) {
    this.basePath = basePath;
    this.commitDirPath = ToyPathDefine.getCommitDirPath(basePath);
  }

  @SneakyThrows
  @Override
  public List<CommitMsg> getCommits(
      CommitType[] commitTypes,
      CommitState[] states,
      ToyVersion from,
      ToyVersion to,
      Long fromTimestamp,
      Long toTimestamp) {

    List<CommitMsg> commits = new ArrayList<>();
    Path compactCommitFilePath =
        new Path(commitDirPath, CompactCommitFile.COMPACT_COMMIT_FILE_NAME);
    FileSystem fileSystem = this.basePath.getFileSystem(new Configuration());

    if (fileSystem.exists(compactCommitFilePath)) {
      CompactCommitFile compactCommitFile = new CompactCommitFile(compactCommitFilePath);
      List<CommitMsg> read = compactCommitFile.read();
      commits.addAll(read);
    }
    FileStatus[] fileStatuses = fileSystem.listStatus(commitDirPath, new CommitFilePathFilter());
    for (FileStatus fileStatus : fileStatuses) {
      try (InputStream in = fileSystem.open(fileStatus.getPath())) {
        Properties p = new Properties();
        p.load(in);
        CommitMsg msg = deserailize(fileStatus.getPath(), p);
        commits.add(msg);
      }
    }
    CommitFilter filter =
        new CommitFilter(commitTypes, states, from, to, fromTimestamp, toTimestamp);
    commits.stream().filter(filter).sorted().collect(Collectors.toList());
    return commits;
  }

  public static class CommitFilter implements Predicate<CommitMsg> {
    CommitType[] commitTypes;
    CommitState[] states;
    ToyVersion from;
    ToyVersion to;
    Long fromTimestamp;
    Long toTimestamp;

    public CommitFilter(
        CommitType[] commitTypes,
        CommitState[] states,
        ToyVersion from,
        ToyVersion to,
        Long fromTimestamp,
        Long toTimestamp) {
      this.commitTypes = commitTypes;
      this.states = states;
      this.from = from;
      this.to = to;
      this.fromTimestamp = fromTimestamp;
      this.toTimestamp = toTimestamp;
    }

    @Override
    public boolean test(CommitMsg commitMsg) {
      if (commitTypes != null && commitTypes.length > 0) {
        if (!ArrayUtils.contains(commitTypes, commitMsg.getCommitType())) {
          return false;
        }
      }

      if (states != null && states.length > 0) {
        if (!ArrayUtils.contains(states, commitMsg.getCommitState())) {
          return false;
        }
      }

      if (from != null && commitMsg.getVersion().compareTo(from) < 0) {
        return false;
      }
      if (to != null && commitMsg.getVersion().compareTo(to) > 0) {
        return false;
      }
      if (fromTimestamp != null && commitMsg.getTimestamp() < fromTimestamp) {
        return false;
      }
      if (toTimestamp != null && commitMsg.getTimestamp() > toTimestamp) {
        return false;
      }
      return true;
    }
  }

  private CommitMsg deserailize(Path commitFilePath, Properties p) {
    String extension = FilenameUtils.getExtension(commitFilePath.getName());
    CommitState state = CommitState.valueOf(extension);

    CommitMsg commitMsg = CommitMsg.deserailize(p);
    commitMsg.setCommitState(state);

    return commitMsg;
  }

  static class CommitFilePathFilter implements PathFilter {

    @Override
    public boolean accept(Path path) {
      Stream<String> collect = Arrays.stream(CommitState.values()).map(Enum::name);
      return collect.anyMatch(p -> path.getName().endsWith(p));
    }
  }
}
